package cn.iocoder.yudao.module.forward.mqtt;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.validation.annotation.Validated;

@Slf4j
@Validated
public class MqttConnect implements MqttCallbackExtended {
    private MqttClient mqttClient;
    private MqttConnectOptions options;
    private MqttInfo mMqttInfo;
    private IMqttCallBack mMqttCallback;
    private volatile int status = MqttConstants.STATUS_DISCONNECTED;

    public MqttConnect(MqttInfo mqttInfo, IMqttCallBack mqttCallback) {
        mMqttInfo = mqttInfo;
        mMqttCallback = mqttCallback;
    }

    public void connect() throws Exception {
        String host = "tcp://" + mMqttInfo.getIp() + ":" + mMqttInfo.getPort();
//        mqttClient = new MqttClient(host, mMqttInfo.getClientId() + "_" + System.currentTimeMillis(), new MemoryPersistence());
        mqttClient = new MqttClient(host, mMqttInfo.getClientId(), new MemoryPersistence());
        options = createMqttConnectOptions();
        mqttClient.setCallback(this);
        mqttClient.connect(options);
    }

    private MqttConnectOptions createMqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(mMqttInfo.getUsername());
        options.setPassword(mMqttInfo.getPwd().toCharArray());
        // 设置是否自动重连
        options.setAutomaticReconnect(true);
        // false 保持会话不被清理自动重连后才能收到订阅的主题消息（包括离线时发布的消息）
        options.setCleanSession(true);
        if (mMqttInfo.getTimeout() != null) {
            options.setConnectionTimeout(mMqttInfo.getTimeout());
        }
        options.setKeepAliveInterval(mMqttInfo.getKeepAliveInterval());
        return options;
    }

    public String publish(String topic, Object obj, int qos) throws Exception {
        String json = null;
        if (obj instanceof String) {
            json = (String) obj;
        } else {
            json = JSONObject.toJSONString(obj);
        }
        log.info("{}发送消息, topic[{}] : {}", mMqttInfo.getClientId(), topic, json);
        mqttClient.publish(topic, json.getBytes(), qos, false);
        return json;
    }

    public void close() throws Exception {
        mqttClient.disconnect();
        mqttClient.close();
        mqttClient = null;
    }

    @Override
    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
        if (mMqttCallback != null) {
            mMqttCallback.messageArrived(mMqttInfo, s, mqttMessage);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        if (mMqttCallback != null) {
            mMqttCallback.deliveryComplete(mMqttInfo, iMqttDeliveryToken);
        }
    }

    @Override
    public void connectionLost(Throwable throwable) {
        status = MqttConstants.STATUS_DISCONNECTED;
        if (mMqttCallback != null) {
            mMqttCallback.connectionLost(mMqttInfo, throwable);
        }
//        try {
//            mqttClient.reconnect();
//        } catch (MqttException e) {
//            e.printStackTrace();
//        }
    }

    @Override
    public void connectComplete(boolean reconnect, String s) {
        status = MqttConstants.STATUS_CONNECTED;
        if (mMqttCallback != null) {
            mMqttCallback.connectComplete(mMqttInfo, reconnect, s);
        }
//        List<String> topics = mMqttInfo.getTopics();
//        if (topics != null && topics.size() > 0) {
//            for (String topic : topics) {
//                try {
//                    mqttClient.subscribe(topic, MqttConstants.MQTT_QOS);
//                } catch (MqttException e) {
//                    e.printStackTrace();
//                }
//                log.info("监听主题({}): {}", mMqttInfo.getName(), topic);
//            }
//        }
    }

    public boolean isConnected() {
        return status == MqttConstants.STATUS_CONNECTED;
    }

    public MqttInfo getMqttInfo() {
        return mMqttInfo;
    }

    public void subscribe(String topic) {
        if (!isConnected()) {
            return;
        }
        boolean ret = false;
        try {
            mqttClient.subscribe(topic);
            ret = true;
        } catch (Exception e) {
        }
        if (mMqttCallback != null) {
            mMqttCallback.onSubscribe(mMqttInfo, ret, topic);
        }
    }

    public void unsubscribe(String topic) {
        if (!isConnected()) {
            return;
        }
        boolean ret = false;
        try {
            mqttClient.unsubscribe(topic);
            ret = true;
        } catch (Exception e) {
        }
        if (mMqttCallback != null) {
            mMqttCallback.onUnsubscribe(mMqttInfo, ret, topic);
        }
    }
}
